welo git mirrors
Commit 1c3f3bf952e1eb883ec1b80119b3a0d84ba0e5dc
Parents : 7c3b042
Author : welo | main nixos computer <empty@empty.com>
Date : 2026-07-08T15:22:57+01:00
checking locking behaviour with sending
Changes
5 files changed, 73 insertions(+), 23 deletions(-)
Diff
diff --git a/src/client.rs b/src/client.rs
index 2f942d1..096dd63 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -260,7 +260,9 @@ async fn dispatch_and_reconnect(
match event {
ProxyEvent::LinkData { data, .. } => {
+ // println!("{:?}",data);
for frame in mux.receive_data(&data) {
+ println!("type {:?} sid: {:?}", frame.frame_type, frame.session_id);
mux.dispatch(frame);
}
}
diff --git a/src/frame.rs b/src/frame.rs
index a8624ed..fb4dc2f 100644
--- a/src/frame.rs
+++ b/src/frame.rs
@@ -9,7 +9,10 @@
use std::fmt;
-use log::info;
+use clap::error::ErrorFormatter;
+use log::{error, info, warn};
+
+use crate::frame::FrameDecodeState::{DecodingFailed, Finished, MoreDataRequired};
// ---------------------------------------------------------------------------
// Frame types and constants
@@ -70,6 +73,12 @@ pub struct Frame {
pub payload: Vec<u8>,
}
+pub enum FrameDecodeState {
+ MoreDataRequired,
+ DecodingFailed,
+ Finished,
+}
+
impl Frame {
pub fn new(frame_type: FrameType, session_id: u32, payload: Vec<u8>) -> Self {
Self {
@@ -92,19 +101,23 @@ impl Frame {
/// Decode a frame from a complete buffer (header + payload).
/// Returns `None` if the buffer is too small or the type is unknown.
- pub fn decode(buf: &[u8]) -> Option<(Self, usize)> {
+ pub fn decode(buf: &[u8]) -> Result<(Self, usize), FrameDecodeState> {
if buf.len() < HDR_SIZE {
- return None;
+ error!("Buf undersized? somehow, probably a bug, packet has been ignored");
+ return Err(Finished);
}
- let frame_type = FrameType::from_u8(buf[0])?;
+ let frame_type = FrameType::from_u8(buf[0]).ok_or(DecodingFailed)?;
let session_id = u32::from_be_bytes([buf[1], buf[2], buf[3], buf[4]]);
let payload_len = u16::from_be_bytes([buf[5], buf[6]]) as usize;
let total = HDR_SIZE + payload_len;
if buf.len() < total {
- return None;
+ return Err(MoreDataRequired);
+ } else if buf.len() < total {
+ // error!("Buf oversized? 100% a bug, ignoring packet hoping that'll fix things");
+ // return Err(DecodingFailed)
}
let payload = buf[HDR_SIZE..total].to_vec();
- Some((
+ Ok((
Self {
frame_type,
session_id,
diff --git a/src/mux.rs b/src/mux.rs
index 9d1cbfd..a16d797 100644
--- a/src/mux.rs
+++ b/src/mux.rs
@@ -15,13 +15,16 @@
//! in rns-rs, causing `NotReady` errors after the first 2 messages.
use std::collections::HashMap;
+use std::error;
use std::sync::{Arc, Mutex};
-use log::{debug, info, warn};
+use log::{debug, error, info, warn};
use rns_core::constants::LINK_MDU;
-use rns_net::{LinkId, RnsNode};
+use rns_core::msgpack::Error;
+use rns_net::{LinkId, LocalServerFactory, RnsNode};
use tokio::sync::mpsc;
+use crate::frame::FrameDecodeState::{DecodingFailed, Finished, MoreDataRequired};
use crate::{Frame, FrameType};
/// Context byte for our link data. We use CONTEXT_NONE (0x00) which routes
@@ -105,16 +108,18 @@ impl MuxHandle {
/// The encoded frame is split into chunks of at most `LINK_MDU` bytes and
/// each chunk is sent as a separate `send_on_link` call with `CONTEXT_NONE`.
/// The receiver reassembles using the frame length header.
+ ///
+ /// Note that we are using .inner.link_id.lock as the guard to prevent
+ /// multiple different sids from sending at the same time and scrambling packets
pub fn send_frame(&self, frame: &Frame) {
- info!("a");
- let link_id = match *self.inner.link_id.lock().unwrap() {
+ let lock = self.inner.link_id.lock();
+ let link_id = match *lock.unwrap() {
Some(id) => id,
None => {
debug!("send_frame: no active link, dropping frame");
return;
}
};
- info!("b");
let encoded = frame.encode();
let node = &self.inner.node;
@@ -153,19 +158,46 @@ impl MuxHandle {
/// Called when `on_link_data` fires. The data may be a partial chunk of a
/// larger frame, so we buffer and try to decode complete frames.
pub fn receive_data(&self, data: &[u8]) -> Vec<Frame> {
+ info!("buf: {:?}", data);
let mut buf = self.inner.recv_buf.lock().unwrap();
buf.extend_from_slice(data);
let mut frames = Vec::new();
loop {
match Frame::decode(&buf) {
- Some((frame, consumed)) => {
+ Ok((frame, consumed)) => {
buf.drain(..consumed);
frames.push(frame);
}
- None => break,
+ Err(err) => {
+ match err {
+ Finished => {
+ // finished decoding all packets basically
+ // though there might still be like 5 bytes left in the buffer
+ break;
+ }
+ MoreDataRequired => {
+ break;
+ // just wait for next packet
+ }
+ DecodingFailed => {
+ // something has gone really wrong, just clear the buffer and hope things
+ // work out.
+ error!("decoding failed for a packet, something really bad is happening {:?}",buf);
+ error!("ignoring packet buffer and hoping that will fix it");
+ buf.drain(..);
+ break;
+ }
+ }
+ },
}
}
+
+ info!("frames {:?}", &frames);
+ if buf.len() > 9000 {
+ warn!("buf thing: {:?}", &buf);
+ assert!(false);
+ }
frames
}
}
diff --git a/src/relay.rs b/src/relay.rs
index 4a96c79..f4a152b 100644
--- a/src/relay.rs
+++ b/src/relay.rs
@@ -1,11 +1,14 @@
//! Bidirectional TCP ↔ RNS relay — used by both client and server sessions.
+//!
+//! Note 1, idk where to put this but the buffer has to be bigger than 64k as v
+//! orginally was 4096 but occasionally there would be packets
use std::net::{Ipv4Addr, ToSocketAddrs};
use std::os::unix::net::SocketAddr;
use std::str::SplitWhitespace;
use std::sync::Arc;
use std::time::Duration;
-
+
use fast_socks5::{new_udp_header, parse_udp_request};
use fast_socks5::util::target_addr::{TargetAddr, ToTargetAddr};
use log::{debug, error, info, warn};
@@ -34,7 +37,7 @@ pub async fn relay_bidirectional_tcp(
// TCP -> RNS
let tcp_to_rns = tokio::spawn(async move {
- let mut buf = [0u8; 4096];
+ let mut buf = [0u8; 65536];
loop {
match tcp_read.read(&mut buf).await {
Ok(0) => break,
@@ -116,7 +119,7 @@ pub async fn relay_bidirectional_udp_client_side(
// UDP -> RNS
let udp_to_rns = tokio::spawn(async move {
- let mut buf = [0u8; 4096];
+ let mut buf = [0u8; 65536];
loop {
let stuff = socket.recv_from(&mut buf).await;
@@ -178,7 +181,7 @@ pub async fn relay_bidirectional_udp_client_side(
let (mut tcp_read, mut _tcp_write) = tcp_stream.into_split();
let break_connection_tcp_check = tokio::spawn(async move {
- let mut buf = [0u8; 4096];
+ let mut buf = [0u8; 65536];
loop {
match tcp_read.read(&mut buf).await {
Ok(0) => {
@@ -222,7 +225,7 @@ pub async fn relay_bidirectional_udp_server_side(
// UDP -> RNS
let udp_to_rns = tokio::spawn(async move {
- let mut buf = [0u8; 4096];
+ let mut buf = [0u8; 65536];
loop {
let stuff = socket.recv_from(&mut buf).await;
@@ -326,7 +329,7 @@ pub async fn relay_forwarded_tcp(
// TCP -> RNS
let tcp_to_rns = tokio::spawn(async move {
- let mut buf = [0u8; 4096];
+ let mut buf = [0u8; 65536];
loop {
match tcp_read.read(&mut buf).await {
Ok(0) => break,
@@ -385,7 +388,7 @@ pub async fn relay_forwarded_udp(
// UDP -> RNS
let tcp_to_rns = tokio::spawn(async move {
- let mut buf = [0u8; 4096];
+ let mut buf = [0u8; 65536];
loop {
match udp_read.read(&mut buf).await {
Ok(0) => break,
diff --git a/src/server.rs b/src/server.rs
index 2083811..7ae84a4 100644
--- a/src/server.rs
+++ b/src/server.rs
@@ -108,12 +108,12 @@ pub async fn run_server(identity_path: Option<&str>, filter_config: FilterConfig
let link_muxes: Arc<Mutex<std::collections::HashMap<LinkId, MuxHandle>>> =
Arc::new(Mutex::new(std::collections::HashMap::new()));
- // Periodic announce task
+ // Periodic announce task. Every 2 hours which is pretty much the minimum
let node_announce = Arc::clone(&node);
let dest_clone = dest.clone();
tokio::spawn(async move {
loop {
- tokio::time::sleep(Duration::from_secs(60)).await;
+ tokio::time::sleep(Duration::from_secs(7200)).await;
let id = Identity::from_private_key(&identity_prv_bytes);
if let Err(e) = node_announce.announce(&dest_clone, &id, None) {
warn!("Failed to send periodic announce: {:?}", e);
@@ -228,7 +228,7 @@ async fn handle_server_session_tcp(
return;
}
};
- info!("{} idk", sid);
+ info!("{} how it went {:?}", sid ,stream);
// Signal success
mux.send(FrameType::ConnectOk, sid, Vec::new());
Served by rngit 1.4.2 - Generated in 0.07s